package com.swak.reactivex.cluster;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import com.swak.Constants;
import com.swak.OS;
import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.impl.ClusterImpl;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.Maps;
import com.swak.utils.StringUtils;

/**
 * 集群
 * 
 * @author 618lf
 */
public interface Cluster {

	Member member();

	Members members();

	void close();

	/**
	 * 创建一个 P2P 集群
	 * 
	 * @param resources 自定义的线程
	 * @param handler   数据处理
	 * @return
	 * @throws RemotingException
	 */
	public static Member join(String clusterName, ExchangeHandler handler) throws RemotingException {
		return Cluster.join(clusterName, UUID.randomUUID().toString(), handler);
	}

	/**
	 * 创建一个 P2P 集群
	 * 
	 * @param resources 自定义的线程
	 * @param handler   数据处理
	 * @return
	 * @throws RemotingException
	 */
	public static Member join(ClusterConfig config, ExchangeHandler handler) throws RemotingException {
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, config.getBindHost());
		parameters.put(Constants.Heartbeat, config.getHeartbeat().toString());
		parameters.put(Constants.Host_Name, config.getHostName());
		parameters.put(Constants.Peer_Id, config.getPeerID());
		parameters.put(Constants.Discovery_Mode_Key, config.getDiscovery().getMode());
		parameters.put(Constants.Discovery_Multicast_Host_Key, config.getDiscovery().getMulticastHost());
		parameters.put(Constants.Discovery_Multicast_Port_Key, config.getDiscovery().getMulticastPort().toString());
		parameters.put(Constants.Discovery_Multicast_Discovery_Times_Key,
				config.getDiscovery().getMulticastDiscoveryTimes().toString());
		parameters.put(Constants.Use_Udp, config.getOpenUDP().toString());
		parameters.put(Constants.Use_Udp_Multicast, config.getOpenMulticast().toString());
		parameters.put(Constants.Multicast_Host_Key, config.getMulticastHost());
		parameters.put(Constants.Multicast_Port_Key, config.getMulticastPort().toString());

		parameters.putAll(config.getParameters());
		URL url = new URL(config.getName(), OS.ip(), OS.randomAvailablePort(), parameters);

		LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, config.getIoSelectThreads(),
				config.getIoWorkerThreads(), config.getName() + ".", false, 2, TimeUnit.SECONDS);
		return Cluster.join(url, resources, handler);
	}

	/**
	 * 创建一个 P2P 集群
	 * 
	 * @param resources 自定义的线程
	 * @param handler   数据处理
	 * @return
	 * @throws RemotingException
	 */
	public static Member join(String clusterName, String id, ExchangeHandler handler) throws RemotingException {
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, Constants.ANYHOST);
		parameters.put(Constants.Heartbeat, String.valueOf(Constants.Default_Heartbeat));
		parameters.put(Constants.Host_Name, OS.hostName());
		parameters.put(Constants.Peer_Id, StringUtils.defaultIfBlank(id, UUID.randomUUID().toString()));
		parameters.put(Constants.Discovery_Multicast_Host_Key, Constants.Multicast_Host);
		parameters.put(Constants.Discovery_Multicast_Port_Key, Constants.Multicast_Port.toString());
		parameters.put(Constants.Discovery_Multicast_Discovery_Times_Key,
				Constants.Multicast_Discovery_Times.toString());
		parameters.put(Constants.Use_Udp, Boolean.TRUE.toString());
		URL url = new URL(clusterName, OS.ip(), OS.randomAvailablePort(), parameters);

		LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 4, clusterName + ".", false, 2,
				TimeUnit.SECONDS);
		return Cluster.join(url, resources, handler);
	}

	/**
	 * 创建一个 P2P 集群
	 * 
	 * @param url       地址
	 * @param resources 自定义的线程
	 * @param handler   数据处理
	 * @return
	 * @throws RemotingException
	 */
	public static Member join(URL url, LoopResources resources, ExchangeHandler handler) throws RemotingException {
		return new ClusterImpl(url, resources, handler).member();
	}
}
